"""
Case Type   : 协议兼容
Case Name   : set_native_password(参数1，参数2，参数3)函数测试
Create At   : 2024.10.11
Owner       : @zhanghuan96
Description :
    1、创建兼容b库并查询插件是否加载成功
    2、创建用户
    3、native认证用户，参数3无参
    4、native认证用户，参数3是空值
    5、再次native认证用户，参数3是空值
    6、再次native认证，新旧密一致
    7、查询系统未使用端口
    8、设置协议兼容guc参数
    9、使用M8.0客户端连接og侧B库并查询og版本
    10、使用M8.0客户端连接og侧B库,协议端口错误
    11、set_native_password函数修改用户密码
    12、使用旧密码连接
    13、使用新密码连接查询当前连接的数据库
    14、再次修改密码，原密码错误
    15、用户不存在使用native认证
    16、使用不存在的用户连接数据库
    17、端口使用--port连接
    18、-h指定的ip地址错误
    19、清理环境
Expect      :
    1、成功，插件查询成功
    2、成功
    3、报错
    4、成功
    5、报错
    6、成功
    7、成功
    8、成功
    9、成功
    10、报错
    11、成功
    12、报错
    13、成功
    14、报错
    15、报错
    16、报错
    17、成功
    18、报错
    19、成功
History     :
"""
import os
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonMySQL import CommonMySQL
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class CompatibilityTest01(unittest.TestCase):

    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.common = Common()
        self.sh_pri = CommonSH('PrimaryDbUser')
        self.pri_node = Node('PrimaryDbUser')
        self.my_node = Node('mysql8')
        self.constant = Constant()
        self.db_name = 'db_protocol_0001'
        self.u_name = 'u_protocol_0001'
        self.u_name_01 = 'u_protocol_0001_01'
        self.pri_node.db_host_error = '1.2.3.4'
        self.pwd = 'Proto_test456'
        self.conf_path = os.path.join(macro.DB_INSTANCE_PATH,
                                      'postgresql.conf')
        self.log.info('查询参数默认值')
        self.default_value1 = self.common.show_param('enable_dolphin_proto')
        self.default_value2 = self.common.show_param('dolphin_server_port')
        self.exp1 = ('ERROR:  function set_native_password(unknown, unknown) '
                     'does not exist')
        self.con_user = (f'-U {self.pri_node.ssh_user} '
                         f'-W {self.pri_node.ssh_password}')
        self.select_sql = 'select version();'

    def test_create_function(self):
        text = '-----step1:创建兼容b库并查询插件;expect:成功-----'
        self.log.info(text)
        sql_cmd1 = (f"drop database if exists {self.db_name};"
                    f"create database {self.db_name}  dbcompatibility ='B';")
        self.log.info(sql_cmd1)
        sql_res1 = self.sh_pri.execut_db_sql(sql_cmd1, dbname='postgres')
        self.log.info(sql_res1)
        self.assertEqual(1,
                         sql_res1.count(self.constant.CREATE_DATABASE_SUCCESS),
                         '兼容B库创建失败')
        sql_cmd1 = "select extname from pg_extension where extname ='dolphin';"
        self.log.info(sql_cmd1)
        sql_res1 = self.sh_pri.execut_db_sql(sql_cmd1, dbname=self.db_name)
        self.log.info(sql_res1)
        self.assertIn('dolphin', sql_res1, '查询失败')

        text = '---step2:创建用户;expect:成功---'
        self.log.info(text)
        sql_cmd2 = (f"drop user if exists {self.u_name};"
                    f"create user {self.u_name} with password "
                    f"'{macro.COMMON_PASSWD}';"
                    f"grant all privileges to {self.u_name};")
        self.log.info(sql_cmd2)
        sql_res2 = self.sh_pri.execut_db_sql(sql_cmd2, sql_type=self.con_user,
                                             dbname=self.db_name)
        self.log.info(sql_res2)
        self.assertIn('CREATE ROLE', sql_res2, '创建失败')
        self.assertIn('ALTER ROLE', sql_res2, '创建失败')

        text = '---step3:native认证用户，参数3无参;expect:合理报错---'
        self.log.info(text)
        sql_cmd3 = (f"select set_native_password('{self.u_name}', "
                    f"'{macro.COMMON_PASSWD}');")
        self.log.info(sql_cmd3)
        sql_res3 = self.sh_pri.execut_db_sql(sql_cmd3, dbname=self.db_name)
        self.log.info(sql_res3)
        self.assertIn(self.exp1, sql_res3, '创建失败')

        text = '---step4:native认证用户，参数3是空值;expect:成功---'
        self.log.info(text)
        sql_cmd4 = (f"select set_native_password('{self.u_name}', "
                    f"'{macro.COMMON_PASSWD}','');")
        self.log.info(sql_cmd4)
        sql_res4 = self.sh_pri.execut_db_sql(sql_cmd4, dbname=self.db_name)
        self.log.info(sql_res4)
        self.assertNotIn('ERROR', sql_res4, '创建失败')
        self.assertIn('(1 row)', sql_res4)

        text = '---step5:再次native认证用户，参数3是空值;expect:合理报错---'
        self.log.info(text)
        sql_cmd5 = (f"select set_native_password('{self.u_name}', "
                    f"'{macro.COMMON_PASSWD}','');")
        self.log.info(sql_cmd5)
        sql_res5 = self.sh_pri.execut_db_sql(sql_cmd5, dbname=self.db_name)
        self.log.info(sql_res5)
        self.assertIn('ERROR:  Set authentication method verification '
                      'failed', sql_res5, '创建失败')

        text = '---step6:再次native认证，新旧密一致;expect:成功---'
        self.log.info(text)
        sql_cmd6 = (f"select set_native_password('{self.u_name}', "
                    f"'{macro.COMMON_PASSWD}','{macro.COMMON_PASSWD}');")
        self.log.info(sql_cmd6)
        sql_res6 = self.sh_pri.execut_db_sql(sql_cmd6, dbname=self.db_name)
        self.log.info(sql_res6)
        self.assertNotIn('ERROR', sql_res6, '创建失败')
        self.assertIn('(1 row)', sql_res6, '创建失败')

        text = '----step7:查询系统未使用端口;expect:成功----'
        self.log.info(text)
        port = self.common.get_not_used_port(self.pri_node)
        self.log.info(port)
        self.assertNotEqual(0, port, '执行失败:' + text)

        text = '---step8:设置协议兼容guc参数;expect:成功---'
        self.log.info(text)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            'enable_dolphin_proto = on')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'dolphin_server_port ={port}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)
        sed_cmd = (f"sed -i '$a dolphin.default_database_name={self.db_name}' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.assertTrue("Degraded" in status or "Normal" in status)

        text = '---step9:使用M8.0客户端连接og侧B库并查询og版本;expect:成功---'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.select_sql, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertNotIn('ERROR', c_res, '执行失败' + text)
        self.assertIn('openGauss', c_res)

        text = '---step10:使用M8.0客户端连接og侧B库,协议端口错误;expect:报错---'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.select_sql, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port='abc', user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('ERROR', c_res, '创建失败')

        text = '---step11:set_native_password函数修改用户密码;expect:成功---'
        self.log.info(text)
        sql_cmd7 = (f"select set_native_password('{self.u_name}', "
                    f"'{self.pwd}','{macro.COMMON_PASSWD}');")
        self.log.info(sql_cmd7)
        sql_res7 = self.sh_pri.execut_db_sql(sql_cmd7, dbname=self.db_name)
        self.log.info(sql_res7)
        self.assertNotIn('ERROR', sql_res7, '创建失败')
        self.assertIn('(1 row)', sql_res7)

        text = '---step12:使用旧密码连接;expect:报错---'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.select_sql, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('ERROR', c_res)

        text = '-step13:使用新密码连接查询当前数据库版本;expect:成功-'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.select_sql, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=self.pwd)
        self.log.info(c_res)
        self.assertIn('openGauss', c_res)

        text = '---step14:再次修改密码，原密码错误;expect:报错---'
        self.log.info(text)
        sql_cmd8 = (f"select set_native_password('{self.u_name}', "
                    f"'{self.pwd}','{self.pwd}_567');")
        self.log.info(sql_cmd8)
        sql_res8 = self.sh_pri.execut_db_sql(sql_cmd8, dbname=self.db_name)
        self.log.info(sql_res8)
        self.assertIn('ERROR:  Set authentication method verification '
                      'failed', sql_res8, '创建失败')

        text = '---step15:用户不存在使用native认证;expect:报错---'
        self.log.info(text)
        sql_cmd9 = (f"select set_native_password('{self.u_name}_01', "
                    f"'{self.pwd}','{self.pwd}_567');")
        self.log.info(sql_cmd9)
        sql_res9 = self.sh_pri.execut_db_sql(sql_cmd9, dbname=self.db_name)
        self.log.info(sql_res9)
        self.assertIn(f'ERROR:  role "{self.u_name}_01" does not '
                      'exist', sql_res9)

        text = '---step16:使用不存在的用户连接数据库;expect:报错---'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.select_sql, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name_01,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('failed in auth check', c_res)

        text = '-step17:端口使用--port连接;expect:成功-'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.select_sql, dbname=self.db_name,
                                    host=self.pri_node.db_host,
                                    port=port, user=self.u_name,
                                    password=self.pwd)
        self.log.info(c_res)
        self.assertIn('openGauss', c_res)

        text = '-step18:-h指定的ip地址错误;expect:报错-'
        self.log.info(text)
        my_sh = CommonMySQL('mysql8')
        c_res = my_sh.execut_db_sql(sql=self.select_sql, dbname=self.db_name,
                                    host=self.pri_node.db_host_error,
                                    port=port, user=self.u_name,
                                    password=macro.COMMON_PASSWD)
        self.log.info(c_res)
        self.assertIn('ERROR', c_res)

    def tearDown(self):
        text = '-----step19:清理环境;expect:成功-----'
        self.log.info(text)
        drop_cmd1 = f"drop user if exists {self.u_name};"
        self.log.info(drop_cmd1)
        drop_res1 = self.sh_pri.execut_db_sql(drop_cmd1, dbname=self.db_name)
        self.log.info(drop_res1)
        drop_cmd = f"drop database if exists {self.db_name};"
        self.log.info(drop_cmd)
        drop_res = self.sh_pri.execut_db_sql(drop_cmd, dbname='postgres')
        self.log.info(drop_res)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'enable_dolphin_proto = '
                                            f'{self.default_value1}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)
        guc_cmd = self.sh_pri.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'dolphin_server_port ='
                                            f'{self.default_value2}')
        self.log.info(guc_cmd)
        self.assertTrue(guc_cmd)

        sed_cmd = (f"sed -i '/dolphin.default_database_name={self.db_name}/d' "
                   f"{self.conf_path}")
        self.log.info(sed_cmd)
        res = self.common.get_sh_result(self.sh_pri.node, sed_cmd)
        self.log.info(res)
        msg = self.sh_pri.restart_db_cluster()
        self.log.info(msg)
        status = self.sh_pri.get_db_cluster_status()
        self.log.info(status)
        self.assertTrue("Degraded" in status or "Normal" in status)
        self.assertIn('DROP ROLE', drop_res1, '清理环境失败')
        self.assertEqual(1,
                         drop_res.count(self.constant.DROP_DATABASE_SUCCESS),
                         '清理环境失败')
        self.log.info(f'-----{os.path.basename(__file__)} end-----')
